package com.interview.rpc.server;

import com.interview.rpc.codec.RpcDecoder;
import com.interview.rpc.codec.RpcEncoder;
import com.interview.rpc.protocol.Request;
import com.interview.rpc.protocol.Response;
import com.interview.rpc.serialize.JsonSerializer;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * RPC服务器
 * 
 * 使用Netty实现的RPC服务器，负责监听客户端连接并处理RPC请求
 */
public class RpcServer {
    
    // 存储服务接口名与实现类实例的映射关系
    private Map<String, Object> serviceMap = new ConcurrentHashMap<>();
    // 服务器监听端口
    private int port;
    
    public RpcServer(int port) {
        this.port = port;
    }
    
    /**
     * 注册服务
     * @param interfaceName 服务接口名
     * @param serviceBean 服务实现类实例
     */
    public void addService(String interfaceName, Object serviceBean) {
        serviceMap.put(interfaceName, serviceBean);
    }
    
    /**
     * 启动RPC服务器
     * @throws Exception
     */
    public void start() throws Exception {
        // 创建主从Reactor线程组
        // bossGroup负责接收客户端连接
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        // workerGroup负责处理IO事件
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        try {
            // 创建服务器启动引导类
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    // 指定通道类型为NioServerSocketChannel
                    .channel(NioServerSocketChannel.class)
                    // 设置子处理器，用于处理客户端连接后的IO事件
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            // 添加解码器，用于将字节流解码为Request对象
                            pipeline.addLast(new RpcDecoder(Request.class, new JsonSerializer()));
                            // 添加编码器，用于将Response对象编码为字节流
                            pipeline.addLast(new RpcEncoder(Response.class, new JsonSerializer()));
                            // 添加服务器处理器，用于处理RPC请求
                            pipeline.addLast(new RpcServerHandler(serviceMap));
                        }
                    })
                    // 设置TCP连接缓冲区大小
                    .option(ChannelOption.SO_BACKLOG, 128)
                    // 启用TCP心跳检测
                    .childOption(ChannelOption.SO_KEEPALIVE, true);
            
            // 绑定端口并启动服务器，sync()方法会阻塞直到绑定完成
            ChannelFuture future = bootstrap.bind(port).sync();
            System.out.println("RPC服务器启动成功，监听端口: " + port);
            // 等待服务器通道关闭，sync()方法会阻塞直到通道关闭
            future.channel().closeFuture().sync();
        } finally {
            // 优雅关闭线程组
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}